package com.dabai.iot.core.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.dabai.common.core.exception.ServiceException;
import com.dabai.common.core.utils.Func;
import com.dabai.common.redis.utils.RedisUtils;
import com.dabai.common.tenant.helper.TenantHelper;
import com.dabai.iot.core.cache.CacheKey;
import com.dabai.iot.core.convert.ServerSubscribeConvert;
import com.dabai.iot.core.dto.ServerSubscribeInput;
import com.dabai.iot.core.dto.ServerSubscribeQuery;
import com.dabai.iot.core.dto.ServerSubscribeUpdate;
import com.dabai.iot.core.entity.ConsumerGroupEntity;
import com.dabai.iot.core.entity.ProductEntity;
import com.dabai.iot.core.entity.ServerSubscribeConsumerEntity;
import com.dabai.iot.core.entity.ServerSubscribeEntity;
import com.dabai.iot.core.mapper.ServerSubscribeMapper;
import com.dabai.iot.core.service.IConsumerGroupService;
import com.dabai.iot.core.service.IProductService;
import com.dabai.iot.core.service.IServerSubscribeConsumerService;
import com.dabai.iot.core.service.IServerSubscribeService;
import com.dabai.iot.core.vo.ServerSubscribeVO;
import lombok.AllArgsConstructor;
import lombok.NonNull;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.stream.Collectors;

/**
 * 服务端订阅 Service 实现类
 *
 * @author kev1n
 */
@Service
@AllArgsConstructor
public class ServerSubscribeServiceImpl extends ServiceImpl<ServerSubscribeMapper, ServerSubscribeEntity> implements IServerSubscribeService {

    private final IProductService productService;
    private final IConsumerGroupService consumerGroupService;
    private final IServerSubscribeConsumerService serverSubscribeConsumerService;

    @Override
    public boolean addServerSubscribe(ServerSubscribeInput serverSubscribeInput) {
        ProductEntity product = productService.getProductCacheById(serverSubscribeInput.getProductId());
        if (product == null) {
            throw new ServiceException("产品不存在");
        }

        Long count = new LambdaQueryChainWrapper<>(baseMapper)
            .eq(ServerSubscribeEntity::getProductId, product.getId())
            .count();
        if (count > 0) {
            throw new ServiceException("该产品已创建订阅");
        }

        ServerSubscribeEntity serverSubscribe = new ServerSubscribeEntity();
        serverSubscribe.setProductId(product.getId());
        serverSubscribe.setMessageType(serverSubscribeInput.getMessageType());
        return save(serverSubscribe);
    }

    /**
     * 修改
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean updateServerSubscribe(ServerSubscribeUpdate serverSubscribeUpdate) {
        ServerSubscribeEntity serverSubscribe = getById(serverSubscribeUpdate.getId());
        if (serverSubscribe == null) {
            // 服务端订阅不存在
            throw new ServiceException("服务端订阅不存在");
        }

        serverSubscribe.setMessageType(serverSubscribeUpdate.getMessageType());

        boolean result = updateById(serverSubscribe);
        if (result) {
            // 删除缓存
            deleteCacheByProductId(serverSubscribe.getProductId());
        }

        return result;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean deleteServerSubscribe(@NonNull Long id) {
        ServerSubscribeEntity serverSubscribe = getById(id);
        if (serverSubscribe == null) {
            throw new ServiceException("服务端订阅不存在");
        }

        List<ConsumerGroupEntity> consumerGroupList = consumerGroupService.getBySubscribeId(id);
        if (Func.isNotEmpty(consumerGroupList)) {
            throw new ServiceException("有消费组关联该订阅，无法删除");
        }

        boolean result = removeById(id);
        if (result) {
            // 删除缓存
            deleteCacheByProductId(serverSubscribe.getProductId());
        }

        return result;
    }

    @Override
    public IPage<ServerSubscribeVO> getServerSubscribePage(IPage<ServerSubscribeEntity> page, ServerSubscribeQuery subscribeQuery) {
        QueryWrapper<ServerSubscribeEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda()
            .eq(Func.isNotEmpty(subscribeQuery.getProductId()), ServerSubscribeEntity::getProductId, subscribeQuery.getProductId())
            .orderByDesc(ServerSubscribeEntity::getCreateTime);

        IPage<ServerSubscribeEntity> result = page(page, queryWrapper);
        return result.convert(this::resolveServerSubscribeVO);
    }

    @Override
    public List<ServerSubscribeVO> getListByConsumerId(@NonNull Long consumerId) {
        List<Long> serverSubscribeIds = serverSubscribeConsumerService.lambdaQuery()
            .eq(ServerSubscribeConsumerEntity::getConsumerId, consumerId)
            .list()
            .stream()
            .map(ServerSubscribeConsumerEntity::getSubscribeId)
            .toList();

        List<ServerSubscribeEntity> list = listByIds(serverSubscribeIds);
        return list.stream().map(this::resolveServerSubscribeVO).collect(Collectors.toList());
    }

    @Override
    public List<ServerSubscribeVO> selectList(String tenantId) {
        List<ServerSubscribeEntity> list = lambdaQuery().eq(ServerSubscribeEntity::getTenantId, tenantId).list();
        return list.stream().map(this::resolveServerSubscribeVO).collect(Collectors.toList());
    }

    private ServerSubscribeVO resolveServerSubscribeVO(ServerSubscribeEntity entity) {
        if (entity == null) {
            return null;
        }

        ServerSubscribeVO vo = ServerSubscribeConvert.INSTANCE.convertVO(entity);

        // 设置产品名称
        ProductEntity product = productService.getProductCacheById(vo.getProductId());
        if (product != null) {
            vo.setProductName(product.getProductName());
        }

        return vo;
    }

    @Override
    public ServerSubscribeEntity getByProductId(@NonNull Long productId) {
        return lambdaQuery().eq(ServerSubscribeEntity::getProductId, productId).one();
    }

    @Override
    public ServerSubscribeEntity getCacheByProductId(Long productId) {
        String cacheKey = CacheKey.getSubscribeKey(productId);
        ServerSubscribeEntity subscribeCache = TenantHelper.ignore(() -> RedisUtils.getCacheObject(cacheKey));
        if (subscribeCache != null) {
            return subscribeCache;
        }

        final ServerSubscribeEntity serverSubscribe = getByProductId(productId);
        if (serverSubscribe != null) {
            subscribeCache = serverSubscribe;
            TenantHelper.ignore(() -> RedisUtils.setCacheObject(cacheKey, serverSubscribe));
        }
        return subscribeCache;
    }


    // private ======================================

    /**
     * 删除服务端订阅缓存
     */
    private void deleteCacheByProductId(@NonNull Long productId) {
        TenantHelper.ignore(() -> RedisUtils.deleteObject(CacheKey.getSubscribeKey(productId)));
    }

}
